package com.demo.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author wangfeng
 * @Description
 * @Date 2021/11/16
 **/

public class TestConsumer {
	private static Logger logger = LoggerFactory.getLogger(TestConsumer.class);
	public static void main(String[] args) throws Exception {
		// 实例化消息生产者,指定组名
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("wudidabaojian42122");
		// 指定Namesrv地址信息.
		consumer.setNamesrvAddr("47.94.17.35:9876");
		// 订阅Topic
		consumer.subscribe("TopicF", "*");
		//负载均衡模式消费
		consumer.setMessageModel(MessageModel.CLUSTERING);

		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
		consumer.setConsumeMessageBatchMaxSize(10);
		consumer.setConsumeThreadMin(1);
		consumer.setConsumeThreadMax(1);
		consumer.setPullBatchSize(10);
		consumer.setPullInterval(10000);

		List<String> list = new ArrayList<>();

		// 注册回调函数，处理消息
		consumer.registerMessageListener(new MessageListenerOrderly() {
			@Override
			public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
													   ConsumeOrderlyContext context) {
				logger.info("queueId = {},msg size = {}",context.getMessageQueue().getQueueId(),msgs.size());
				for (MessageExt msg : msgs){
					String str = new String(msg.getBody());
					logger.info(str);
					list.add(str);
				}

				/*System.out.printf("%s Receive New Messages: %s %n",
						Thread.currentThread().getName(), msgs);*/
				return ConsumeOrderlyStatus.SUCCESS;
			}
		});
		logger.info(list.toString());
		//启动消息者
		consumer.start();
		System.out.printf("Consumer Started.%n");
	}
}
